package com.shujia.flink.state

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.lang
import scala.collection.mutable
import scala.collection.mutable.ListBuffer

object Demo1CardAvgSpeed {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val carDS: DataStream[String] = env.socketTextStream("master", 8888)

    //解析数据
    val cardAndSpeedDS: DataStream[(Long, Double)] = carDS.map(line => {
      val jsonObj: JSONObject = JSON.parseObject(line)
      val card: Long = jsonObj.getLong("card")
      val speed: Double = jsonObj.getDouble("speed")
      (card, speed)
    })
    /**
     * 实时计算每隔kafkou的平均车速
     */
    //安装卡口分组
    val keyByDS: KeyedStream[(Long, Double), Long] = cardAndSpeedDS.keyBy(_._1)

    val avgSpeedDS: DataStream[(Long, Double)] = keyByDS
      .process(new KeyedProcessFunction[Long, (Long, Double), (Long, Double)] {

        //同一个task中所有的数据共享同一个变量
        //val speeds = new ListBuffer[Double]

        //为一个key保存一个集合
        /**
         * 使用普通scala集合保存中间结果的问题
         * 1、如果任务执行失败，集合中的数据会丢失
         * 2、任务重启之后没有办法获取到之前的计算结果
         */
        val hashMap = new mutable.HashMap[Long, ListBuffer[Double]]()

        /**
         * processElement: 每一条数据执行一次
         *
         * @param value ：一行数据
         * @param ctx   ： 上下文对象
         * @param out   ： 用于将数据发送到下游
         */
        override def processElement(value: (Long, Double),
                                    ctx: KeyedProcessFunction[Long, (Long, Double), (Long, Double)]#Context,
                                    out: Collector[(Long, Double)]): Unit = {
          val (card: Long, speed: Double) = value

          //从hash中取出当前key所有的车速
          val speeds: ListBuffer[Double] = hashMap.getOrElse(card, ListBuffer[Double]())

          //将每一条数据的车速保存到集合中
          speeds += speed

          //计算平均车速
          val avgSpeed: Double = speeds.sum / speeds.length

          //将计算结果发送到下游
          out.collect((card, avgSpeed))

          //将最新的车速保存到hashmap中
          hashMap.put(card, speeds)
        }
      })

    avgSpeedDS.print()

    env.execute()
  }

}
